package cc.iotkit.temporal.es.service;

import cc.iotkit.temporal.es.document.DocDeviceConsumePower;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.QueryBuilders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DataAccessException;
import org.springframework.data.domain.PageRequest;
import org.springframework.data.domain.Sort;
import org.springframework.data.elasticsearch.core.ElasticsearchRestTemplate;
import org.springframework.data.elasticsearch.core.SearchHit;
import org.springframework.data.elasticsearch.core.SearchHits;
import org.springframework.data.elasticsearch.core.mapping.IndexCoordinates;
import org.springframework.data.elasticsearch.core.query.NativeSearchQuery;
import org.springframework.data.elasticsearch.core.query.NativeSearchQueryBuilder;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * 过期能耗数据删除定时任务
 */
@Slf4j
@Component
public class DeviceEnergyRemoveTask {

    private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")
            .withZone(ZoneId.systemDefault());

    @Value("${device.energy.retention-days:180}")
    private int retentionDays;

    @Autowired
    private ElasticsearchRestTemplate template;

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 删除过期(超过六个月)设备能耗(每天1点25分执行)
     */
    @Scheduled(cron = "0 25 1 * * ?")
    public void expiredDeviceEnergyRemove() {
        long now = System.currentTimeMillis();
        long expireTime = now - TimeUnit.DAYS.toMillis(retentionDays);

        IndexCoordinates index = IndexCoordinates.of("device_kilowatt_hour");
        long totalDeleted = 0;
        int pageSize = 1000;
        int page = 0;

        try {
            List<Object> searchAfter = null;
            while (true) {
                NativeSearchQueryBuilder queryBuilder = new NativeSearchQueryBuilder()
                        .withQuery(QueryBuilders.rangeQuery("time").lt(expireTime))
                        .withSort(Sort.by("time").ascending())
                        .withSort(Sort.by("_id").ascending()) // 保证唯一排序
                        .withPageable(PageRequest.of(0, pageSize)); // 这里 page 固定 0

                if (searchAfter != null) {
                    queryBuilder.withSearchAfter(searchAfter);
                }

                NativeSearchQuery query = queryBuilder.build();

                SearchHits<DocDeviceConsumePower> searchHits = template.search(query, DocDeviceConsumePower.class, index);

                if (searchHits.isEmpty()) {
                    break;
                }

                List<String> idsToDelete = searchHits.getSearchHits().stream()
                        .map(SearchHit::getId)
                        .collect(Collectors.toList());

                if (!idsToDelete.isEmpty()) {
                    // 批量删除
                    BulkRequest bulkRequest = new BulkRequest();
                    idsToDelete.forEach(id -> bulkRequest.add(new DeleteRequest(index.getIndexName(), id)));
                    BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                    if (bulkResponse.hasFailures()) {
                        log.warn("批量删除部分失败: {}", bulkResponse.buildFailureMessage());
                    }
                    totalDeleted += idsToDelete.size();
                }

                // 取最后一个 hit 的 sort 值，用作下一次 search_after
                searchAfter = searchHits.getSearchHits()
                        .get(searchHits.getSearchHits().size() - 1)
                        .getSortValues();
            }

            log.info("定时任务: 删除了 {} 条 {} 天前的设备能耗数据 (截止时间: {})",
                    totalDeleted, retentionDays, DATE_FORMATTER.format(Instant.ofEpochMilli(expireTime)));
                    //new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(expireTime)));

        } catch (DataAccessException e) {
            log.error("数据库访问异常，删除过期设备能耗失败", e);
        } catch (IOException e) {
            log.error("Elasticsearch IO 异常，删除过期设备能耗失败", e);
        } catch (Exception e) {
            log.error("删除过期设备能耗失败", e);
        }
    }

    public void expiredDeviceEnergyRemoveOld() {
        long now = System.currentTimeMillis();
        long expireTime = now - TimeUnit.DAYS.toMillis(retentionDays);

        IndexCoordinates index = IndexCoordinates.of("device_kilowatt_hour");
        long totalDeleted = 0;
        int pageSize = 1000;
        int page = 0;

        try {
            SearchHits<DocDeviceConsumePower> searchHits;
            do {
                // 构造带分页的查询条件
                NativeSearchQuery query = new NativeSearchQueryBuilder()
                        .withQuery(QueryBuilders.rangeQuery("time").lt(expireTime))
                        .withPageable(PageRequest.of(page++, pageSize))
                        .build();

                searchHits = template.search(query, DocDeviceConsumePower.class, index);
                List<String> idsToDelete = searchHits.getSearchHits().stream()
                        .map(SearchHit::getId)
                        .collect(Collectors.toList());

                if (!idsToDelete.isEmpty()) {
                    // 批量删除
                    BulkRequest bulkRequest = new BulkRequest();
                    for (String id : idsToDelete) {
                        DeleteRequest deleteRequest = new DeleteRequest(index.getIndexName(), id);
                        bulkRequest.add(deleteRequest);
                    }
                    BulkResponse bulkResponse = restHighLevelClient.bulk(bulkRequest, RequestOptions.DEFAULT);
                    if (bulkResponse.hasFailures()) {
                        log.warn("批量删除部分失败: {}", bulkResponse.buildFailureMessage());
                    }

                    totalDeleted += idsToDelete.size();
                }

            } while (searchHits.hasSearchHits());

            // 最终刷新索引
            template.indexOps(index).refresh();

            log.info("定时任务: 删除了 {} 条 {} 天前的设备能耗数据 (截止时间: {})",
                    totalDeleted, retentionDays, DATE_FORMATTER.format(Instant.ofEpochMilli(expireTime)));
            //new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(expireTime)));

        } catch (DataAccessException e) {
            log.error("数据库访问异常，删除过期设备能耗失败", e);
        } catch (IOException e) {
            log.error("Elasticsearch IO 异常，删除过期设备能耗失败", e);
        } catch (Exception e) {
            log.error("删除过期设备能耗失败", e);
        }
    }

}
